#ifndef ACE_TCP_CONNECTOR_IMPL_H
#define ACE_TCP_CONNECTOR_IMPL_H

#include "tcp_session.h"

#include "ace/Acceptor.h"
#include "ace/Reactor.h"
#include "ace/TP_Reactor.h"
#include "ace/SOCK_Stream.h"
#include "ace/SOCK_Connector.h"
#include "ace/Connector.h"
#include "ace/INET_Addr.h"

#include "tcp_message_context.h"

#include <string>
#include <boost/bind.hpp>

namespace vpm
{
	namespace mina
	{
		namespace detail
		{			
            template <class Session= TcpSession,class SessionConfig = TcpSessionConfig > 
			class AceTcpConnectorImpl : public ACE_Event_Handler 
			{
			public:
				enum 
				{
					TCP_DATA_BUFFER_LENGTH = 8192
				};
				AceTcpConnectorImpl(TcpConnectorConfig& config ,IoHandler& handler)
                    :m_connectorConfig(config),m_ioHandler(handler),m_thrdGrpId(-1) 
				{
					m_pReactor.reset(new ACE_Reactor( new ACE_TP_Reactor(),1)); 
					this->reactor(m_pReactor.get());

					m_tcpSession.reset(new Session(config.get_session_config()));
					IoSessionAccessor::attach_writter(*m_tcpSession,boost::bind(&AceTcpConnectorImpl::send_data,this,_1));
					m_ioHandler.sesion_created(*m_tcpSession);
				}

				~AceTcpConnectorImpl()
				{
					this->reactor(NULL);
				}

				int connect(int port, const std::string & host= "0.0.0.0")
				{
					m_thrdGrpId = ACE_Thread_Manager::instance()->spawn_n(1,AceTcpConnectorImpl::tcp_connector_reactor_event_loop,m_pReactor.get()); 
					if (m_thrdGrpId == -1 )
					{
						return -1; 
					}

					ACE_SOCK_Connector sockConnector;
					ACE_Time_Value wait(0,500); 

					int ret = sockConnector.connect(m_sockStream,ACE_INET_Addr(port,host.c_str()), &wait); 
					if (ret == 0 )
					{				
						ret = this->reactor()->register_handler(m_sockStream.get_handle() ,this, ACE_Event_Handler::READ_MASK );				
					}

					if (ret == 0 )
					{
						//session opened
						m_ioHandler.session_opened(*m_tcpSession);
					} 
					return ret; 
				}

				int disconnect()
				{		
					if (m_thrdGrpId != -1 )
					{
						m_thrdGrpId = -1; 
						m_sockStream.close(); 
						m_pReactor->remove_handler(this,ACE_Event_Handler::READ_MASK); 
						m_pReactor->end_reactor_event_loop(); 			
						ACE_Thread_Manager::instance()->wait_grp(m_thrdGrpId);			
					}
					return 0; 		
				}

				int server_port()
				{
					return m_serverPort; 
				}

				const std::string server_host()
				{
					return m_serverHost;
				}

				int send_data(MessageContext & msg)
				{
					TcpMessageContext & tcpMsg = (TcpMessageContext&)msg; 
					return this->send_data(tcpMsg.buffer,tcpMsg.length);		
				}

				int send_data(const char * pData, int dataLen)
				{
					int ret = m_sockStream.send(pData,dataLen);
					if (ret < 0)
					{			
						return -1; 
					}
					return ret; 
				}

				int recv_data(MessageContext & msg)
				{
					TcpMessageContext & tcpMsg = (TcpMessageContext&) msg;
					return this->recv_data(tcpMsg.buffer,tcpMsg.length);
				}

				virtual int recv_data(char * pBuf, int bufLen)
				{
					return m_sockStream.recv(pBuf,bufLen);
				}

				virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE)
				{
					int recvLen = this->m_sockStream.recv(m_dataBuf  , TCP_DATA_BUFFER_LENGTH  ); 
					if (recvLen < 0 ) 
					{
						ACE_DEBUG ((LM_DEBUG,ACE_TEXT ("Connection closed\n")));   
						return -1;
					}						
					return 0; 
				}

				virtual int handle_timeout(const ACE_Time_Value &expire, const void *act = NULL)
				{
					return 0; 
				}

				virtual int handle_close(ACE_HANDLE handle,ACE_Reactor_Mask close_mask)
				{
					return 0; 
				}

				static ACE_THR_FUNC_RETURN tcp_connector_reactor_event_loop(void * pArg)
				{
					assert(pArg != NULL); 
					ACE_Reactor * pReactor = (ACE_Reactor*) pArg; 
					pReactor->owner(ACE_OS::thr_self()); 
					pReactor->run_reactor_event_loop(); 
					ACE_DEBUG((LM_DEBUG,ACE_TEXT("Quit the reactor event loop\n"))); 
					return 0; 
				}
			private:
				ACE_SOCK_Stream m_sockStream;
				std::auto_ptr<ACE_Reactor> m_pReactor; 

				char m_dataBuf[TCP_DATA_BUFFER_LENGTH]; 
	
				int m_serverPort; 
				std::string m_serverHost; 
				TcpConnectorConfig& m_connectorConfig; 				
				IoHandler & m_ioHandler; 	
				int m_thrdGrpId; 
				std::auto_ptr<IoSession> m_tcpSession; 
			};

#define TcpConnectorImpl AceTcpConnectorImpl
		}
	}
}

#endif //
  
